#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <errno.h>
#include <iostream>

#include "MetadataSvr.h"
#include "ProtoHelper.h"
#include "MetadataSvrCommon.h"
#include "MetadataSvrProtoHelper.h"
#include "HashHelper.h"

#define DS_STATUS_OK 0
#define DS_NOTIFY_DATA_FILE_INFO_REASON_FILE_OP 0
#define DS_NOTIFY_DATA_FILE_INFO_REASON_REPLICATE 1

using std::cout;
using std::endl;

LoggerPtr MetadataSvr::_logger = Logger::getLogger("MetadataSvr");

int MetadataSvr::init(const char * log_conf_file, int my_mds_id, const char * cs_ip, unsigned short cs_port, \
				int cs_conn_pool_size, int cs_socket_conn_timeout, int cs_socket_send_timeout, \
				int cs_socket_recv_timeout, int cs_socket_max_response_size) {
	if (NULL == log_conf_file || my_mds_id < 0 || NULL == cs_ip) {
		LOG4CXX_ERROR(_logger, "init error, invalid args");
		return -1;
	}
	srand(time(NULL));
	_sfw.init_logger(log_conf_file);

	int ret = _csc.init(cs_ip, cs_port, cs_conn_pool_size, cs_socket_conn_timeout, cs_socket_send_timeout, \
                                        cs_socket_recv_timeout, cs_socket_max_response_size);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", cnf svr init error, ret:" << ret);
                return -2;
        }

	list<MDS_CNF> l_mds;
        ret = _csc.get_all_mds_cnf(l_mds);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get all mds cnf, ret:" << ret);
                return -3;
        }
        bool found = false;
        for(list<MDS_CNF>::const_iterator iter = l_mds.begin(); l_mds.end() != iter; iter++) {
                if (iter->id == my_mds_id) {
                        _listen_ip = iter->ip;
                        _listen_port = iter->port;
                        _thread_num = iter->thread_num;
                        _log_conf_file = log_conf_file;
                        _socket_idle_timeout = iter->socket_idle_timeout;
                        _socket_send_timeout = iter->socket_send_timeout;
                        _socket_recv_timeout = iter->socket_recv_timeout;
                        _max_request_size = iter->socket_max_request_size;
                        _root_data_path = iter->root_data_path;
                        _my_mds_id = my_mds_id;

                        ret = _sfw.init(iter->thread_num, handle_socket_input_func, this, iter->ip.c_str(), iter->port, \
                                        iter->socket_idle_timeout, NULL, NULL);
                        found = true;
                        if (ret < 0) {
                                LOG4CXX_ERROR(_logger, "sfw init error, ret:" << ret);
                                return -4;
                        }

                }
        }
	if (!found) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get my id from conf server");
                return -5;
        }

	pthread_mutex_init(&_replicate_tasks_mutex, NULL);
	pthread_cond_init(&_replicate_tasks_cond,  NULL);
	for (int i = 0; i < WRITE_MD_FILE_MUTEX_NUM; i++) {
		pthread_mutex_init(&_write_md_file_mutexes[i], NULL);
	}
	ret = _dscg.init(&_csc);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr client group init error, ret:" << ret);
                return -6;
        }
	_replicate_factor = 2;
	_min_modify_interval_before_replicate = 10;
	pthread_create(&_replicate_notify_thread, NULL, notify_replicate_thread_routine, this);

	LOG4CXX_INFO(_logger, "init MetadataSvr OK, listen_ip:" << _listen_ip << ", listen_port:" << _listen_port << \
				", thread_num:" << \
				_thread_num << ", log_conf_file:" << _log_conf_file << ", socket_idle_timeout:" << \
				_socket_idle_timeout << ", socket_send_timeout:" << _socket_send_timeout << \
				", socket_recv_timeout:" << _socket_recv_timeout << ", max_request_size:" << _max_request_size << \
				", root_data_path:" << _root_data_path);
	return 1;
}

int MetadataSvr::start() {
	return _sfw.start_server();
}

int MetadataSvr::handle_socket_input_func(int fd, void * arg) {
	MetadataSvr * ns = (MetadataSvr *)arg;
        json_object * j;
        int cmd_id;
        string request;
        int ret = recv_and_parse_request(fd, ns->_socket_recv_timeout, ns->_max_request_size, j, cmd_id, request);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, "recv and parse request error, ret:" << ret);
                return -1;
        }
        string response;
        ns->dispatch_request(cmd_id, j, response);
        json_object_put(j);
        if (response.size() < 1) response = build_error_response(MDS_ERR_UNKNOWN_ERR);
        ret = send_cmd(fd, ns->_socket_send_timeout, response.data(), response.size());
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, "reply cmd error, ret:" << ret << ", fd:" << fd << ", request:" << request << \
                                        ", response:" << response);
                return -2;
        }
        return 2;
}

int MetadataSvr::dispatch_request(int cmd_id, json_object * j, string & response) {
	int ret = -1;
        switch (cmd_id) {
                case MDS_CMD_CREATE_MD_FILE:
                        ret = handle_create_md_file(j, response);
                        break;
                case MDS_CMD_DELETE_MD_FILE:
                        ret = handle_delete_md_file(j, response);
                        break;
		case MDS_CMD_DATA_SVR_NOTIFY_DATA_FILE_INFO:
			ret = handle_data_svr_notify_data_file_info(j, response);
			break;
		case MDS_CMD_CLIENT_NOTIFY_CLIENT_DATA_INFO:
			ret = handle_client_notify_client_data_info(j, response);
			break;
		case MDS_CMD_GET_MD_FILE_INFO:
			ret = handle_get_md_file_info(j, response);
			break;
		case MDS_CMD_ALLOC_DS_4_NEW_FILE:
			ret = handle_alloc_ds_4_new_file(j, response);
			break;
                default:
                        LOG4CXX_ERROR(_logger, "unknown command, cmd_id:" << cmd_id);
                        ret = -100;
        }
        return ret;
}

int MetadataSvr::handle_create_md_file(json_object * j, string & response) {
	json_object * j_dir_id = json_object_object_get(j, "DIR_ID");	
	json_object * j_file_id = json_object_object_get(j, "FILE_ID");
	if (is_error(j_dir_id) || is_error(j_file_id)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -1;
        }
	int32_t dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	if (dir_id < 0 || file_id < 1) {
	        LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args error:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -2;
        }
	string file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	struct stat s;
	int ret = stat(file_path.c_str(), &s);
	if (0 == ret)  {
		LOG4CXX_ERROR(_logger, __FUNCTION__ <<",  file already exist, file:" << file_path);
		response = build_error_response(MDS_ERR_FILE_ALREADY_EXIST);
		return -3;
	}
	FILE * fp = fopen(file_path.c_str(), "w");
	if (NULL == fp) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ <<",  fopen error, file:" << file_path << ", errno:" << \
				errno);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -4;
	}
	fclose(fp);
	LOG4CXX_INFO(_logger, "file create ok, file:" << file_path); 
	response = build_ok_response();
	return 1;
}

int MetadataSvr::handle_delete_md_file(json_object * j, string & response) {
	json_object * j_dir_id = json_object_object_get(j, "DIR_ID");	
	json_object * j_file_id = json_object_object_get(j, "FILE_ID");
	if (is_error(j_dir_id) || is_error(j_file_id)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -1;
        }
	int32_t dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	if (dir_id < 0 || file_id < 1) {
	        LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args error:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -2;
        }
	string file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	
	struct stat s;
	int ret = stat(file_path.c_str(), &s);
	if (0 != ret)  {
		if (ENOENT == errno) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", file not exist, file:" << file_path);
			response = build_error_response(MDS_ERR_FILE_NOT_EXIST);
			return -3;
		}
		else {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ",  stat file error:" << file_path << ", stat ret:" << \
					ret << ", errno:" << errno);
			response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
			return -4;
		}
	}
	if (!S_ISREG(s.st_mode)) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ <<  ", try to delete a file but the input file path is not a file:" << \
					file_path);
			response = build_error_response(MDS_ERR_TARGET_NOT_A_FILE);
			return -5;
	}
	ret = unlink(file_path.c_str());
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", unlink file error, file:" << file_path << ", ret:" << ret << \
				", errno:" << errno);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -6;
	}
	LOG4CXX_INFO(_logger, "delete file ok, file:" << file_path);
	response = build_ok_response();
	return 1;
}

int MetadataSvr::handle_data_svr_notify_data_file_info(json_object * j, string & response) {
	json_object * j_dir_id = json_object_object_get(j, "DIR_ID");	
	json_object * j_file_id = json_object_object_get(j, "FILE_ID");
	json_object * j_crcs = json_object_object_get(j, "CRC");
	json_object * j_file_size = json_object_object_get(j, "FILE_SIZE");
	json_object * j_data_svr_id = json_object_object_get(j, "DATA_SVR_ID");
	json_object * j_delete_flag = json_object_object_get(j, "DELETE_FLAG");
	json_object * j_notify_reason = json_object_object_get(j, "NOTIFY_REASON");
	if (NULL == j_dir_id || NULL == j_file_id || NULL ==  j_crcs || NULL == j_file_size || NULL == j_data_svr_id || \
		NULL == j_delete_flag) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -1;
        }
	int32_t dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	int64_t file_size = json_object_get_int(j_file_size);
	int data_svr_id = json_object_get_int(j_data_svr_id);
	int delete_flag = json_object_get_int(j_delete_flag);
	int notify_reason = json_object_get_int(j_notify_reason);
	if (dir_id < 0 || file_id < 1) {
	        LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -2;
        }
	vector<uint32_t> crcs;
	for(int i=0; i < json_object_array_length(j_crcs); i++) {
		json_object * j_t = json_object_array_get_idx(j_crcs, i);
		crcs.push_back(json_object_get_int(j_t));
	}
	string file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	acquire_write_md_file_lock(dir_id, file_id);
	int ret = update_md_file_info_by_data_svr_notify(file_path.c_str(), data_svr_id, crcs, file_size, delete_flag);
	release_write_md_file_lock(dir_id, file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", update_md_file_info_by_data_svr_notify error, file:" << file_path << \
			", data_svr_id:" << data_svr_id << ", file_size:" << file_size << ", delete_flag:" << delete_flag << \
			", ret:" << ret);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	if (DS_NOTIFY_DATA_FILE_INFO_REASON_REPLICATE == notify_reason) {
		REPLICATE_TASK t;
		t.dir_id = dir_id;
		t.file_id = file_id;
		erase_replicate_task_from_waiting_response_queue(t);
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, request:" << json_object_to_json_string(j));
	response = build_ok_response();
	return 1;
}


int MetadataSvr::handle_client_notify_client_data_info(json_object * j, string & response) {
	json_object * j_dir_id = json_object_object_get(j, "DIR_ID");	
	json_object * j_file_id = json_object_object_get(j, "FILE_ID");
	json_object * j_crcs = json_object_object_get(j, "CRC");
	if (is_error(j_dir_id) || is_error(j_file_id) || is_error(j_crcs)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -1;
        }
	int32_t dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	if (dir_id < 0 || file_id < 1) {
	        LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args error:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -2;
        }
	vector<uint32_t> crcs;
	for(int i=0; i < json_object_array_length(j_crcs); i++) {
		json_object * j_t = json_object_array_get_idx(j_crcs, i);
		crcs.push_back(json_object_get_int(j_t));
	}
	string file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	acquire_write_md_file_lock(dir_id, file_id);
	int ret = update_md_file_info_by_client_notify(file_path.c_str(), crcs);
	release_write_md_file_lock(dir_id, file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", update_md_file_info_by_client_notify error, file:" << file_path << \
			 ", ret:" << ret);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, request:" << json_object_to_json_string(j));
	response = build_ok_response();
	return 1;
}

int MetadataSvr::handle_get_md_file_info(json_object * j, string & response) {
	json_object * j_dir_id = json_object_object_get(j, "DIR_ID");	
	json_object * j_file_id = json_object_object_get(j, "FILE_ID");
	if (is_error(j_dir_id) || is_error(j_file_id)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args missing:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -1;
        }
	int32_t dir_id = json_object_get_int(j_dir_id);
	int64_t file_id = json_object_get_int(j_file_id);
	if (dir_id < 0 || file_id < 1) {
	        LOG4CXX_ERROR(_logger, __FUNCTION__ << ", args error:" << json_object_to_json_string(j));
		response = build_error_response(MDS_ERR_INVALID_ARGS);
		return -2;
        }
	string file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	/*
	struct stat s;
	int ret = stat(file_path.c_str(), &s);
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << file_path << ", errno:" << errno);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	FILE * fp = fopen(file_path.c_str(), "r");
	if (NULL == fp) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fopen error, file:" << file_path << ", errno:" << errno);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -4;
	}
	char * buf = new char[s.st_size + 1];
	bzero(buf, s.st_size + 1);
	int r = fread(buf, s.st_size, 1, fp);
	fclose(fp);
	if (1 != r) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fread error, file:" << file_path << ", errno:" << errno);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -5;
	}
	*/
	MD_FILE_INFO info;
	acquire_write_md_file_lock(dir_id, file_id);
	int ret = get_md_file_info(file_path.c_str(), info);
	release_write_md_file_lock(dir_id, file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get_md_file_info error, file:" << file_path << ", ret:" << ret);
		response = build_error_response(MDS_ERR_OS_FILE_OP_ERR);
		return -3;
	}
	
	//response = mds_build_get_md_file_info_response(buf);
	response = mds_build_get_md_file_info_response(info);
	check_replicate(dir_id, file_id, info);
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, request:" << json_object_to_json_string(j) << ", response:" << response);
	return 1;
}

int MetadataSvr::handle_alloc_ds_4_new_file(json_object * j, string & response) {
	list<DS_CNF> l_ds;
        int ret = _csc.get_all_ds_cnf(l_ds);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get all ds cnf error, ret:" << ret);
                response = build_error_response(MDS_ERR_OUTER_SERVICE_ERR);
                return -1;
        }
        int r = rand();
        int total_weight = 0;
        for (list<DS_CNF>::const_iterator iter = l_ds.begin(); l_ds.end() != iter; iter++) {
                if (iter->write_weight <= 0 || DS_STATUS_OK != iter->status || iter->mds_id != _my_mds_id) continue;
                total_weight += iter->write_weight;
        }
        if (total_weight < 1) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", total weight is 0" << ret);
                response = build_error_response(MDS_ERR_OUTER_SERVICE_ERR);
                return -2;
        }
        r = r % total_weight;
        int t = 0;
        int ds_id = -1;
        for (list<DS_CNF>::const_iterator iter = l_ds.begin(); l_ds.end() != iter; iter++) {
                if (iter->write_weight <= 0 || DS_STATUS_OK != iter->status || iter->mds_id != _my_mds_id) continue;
                t += iter->write_weight;
                if (t >= r) { ds_id = iter->id; break; }
        }
	int32_t dir_id = rand();
	int32_t file_id = time(NULL);
        response = mds_build_alloc_ds_4_new_file_response(ds_id, dir_id, file_id);
        LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, response:" << response);
        return 1;
}


int MetadataSvr::get_md_file_info(const char * file_path, MD_FILE_INFO & info) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << " error, null file path");
		return -1;
	}
	struct stat s;
	int ret = stat(file_path, &s);
	if (0 != ret)  {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", stat error, file:" << file_path << ", errno:" << errno);
		return -2;
	}
	if (s.st_size < 1) return 1;
	FILE * fp = fopen(file_path, "r");
	if (NULL == fp) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fopen eror, file:" << file_path << ", errno:" << errno);
		return -2;
	}
	char * buf = new char[s.st_size + 1];
	bzero(buf, s.st_size + 1);
	int r = fread(buf, s.st_size, 1, fp);
	fclose(fp);
	if (1 != r) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fread error, file:" << file_path << ", ret:" << r << ", errno:" << errno);
		delete[] buf;
	}
	json_object * j = json_tokener_parse(buf);
	delete[] buf;
        if (is_error(j)) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", parse json string err:" << buf); 
                return -3;
        }
	json_object * j_client_md_record = json_object_object_get(j, "CLIENT_MD_RECORD");
	if (NULL != j_client_md_record && !is_error(j_client_md_record)) {
		json_object * j_client_crcs = json_object_object_get(j_client_md_record, "CLIENT_CRC");
		if (NULL != j_client_crcs && !is_error(j_client_crcs)) {
			for(int i=0; i < json_object_array_length(j_client_crcs); i++) {
				json_object * j_t = json_object_array_get_idx(j_client_crcs, i);
				info.client_md_record.crcs.push_back(json_object_get_int(j_t));
			}

			
		}
	}
	json_object * j_data_svr_md_records = json_object_object_get(j, "DATA_SVR_MD_RECORDS");
	if (NULL != j_data_svr_md_records && !is_error(j_data_svr_md_records)) {
			for(int i=0; i < json_object_array_length(j_data_svr_md_records); i++) {
				json_object * j_data_svr_md_record = json_object_array_get_idx(j_data_svr_md_records, i);
				json_object * j_data_svr_id = json_object_object_get(j_data_svr_md_record, "DATA_SVR_ID");
				json_object * j_file_size = json_object_object_get(j_data_svr_md_record, "FILE_SIZE");
				json_object * j_crcs = json_object_object_get(j_data_svr_md_record, "CRC");
				int data_svr_id = json_object_get_int(j_data_svr_id);
				int64_t file_size = json_object_get_int(j_file_size);
				vector<uint32_t> crcs;
				for (int j = 0; j < json_object_array_length(j_crcs); j++) {
					json_object * j_t = json_object_array_get_idx(j_crcs, j);
					crcs.push_back(json_object_get_int(j_t));
				}
				DATA_SVR_MD_RECORD r;
				r.crcs = crcs;
				r.file_size = file_size;
				info.data_svr_md_records.insert(map<int, DATA_SVR_MD_RECORD>::value_type(data_svr_id, r));
			}
	}
	json_object_put(j);
	return 2;
}

int MetadataSvr::write_md_file_info(const char * file_path, MD_FILE_INFO  info) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << " error, null file path");
		return -1;
	}
	json_object * j = json_object_new_object();
	json_object * j_client_md_record = json_object_new_object();
	json_object * j_data_svr_md_records = json_object_new_array();
	json_object * j_client_crcs = json_object_new_array();
	for(vector<uint32_t>::const_iterator iter = info.client_md_record.crcs.begin(); \
		info.client_md_record.crcs.end() != iter; iter++) {
		json_object * j_crc = json_object_new_int(*iter);
		json_object_array_add(j_client_crcs, j_crc);
	}
	json_object_object_add(j_client_md_record, "CLIENT_CRC", j_client_crcs);
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = info.data_svr_md_records.begin(); \
			info.data_svr_md_records.end() != iter; iter++) {
		json_object * j_data_svr_md_record = json_object_new_object();
		json_object * j_data_svr_id = json_object_new_int(iter->first);
		json_object * j_file_size = json_object_new_int(iter->second.file_size);
		json_object * j_crcs = json_object_new_array();
		for (vector<uint32_t>::const_iterator iter1 = iter->second.crcs.begin(); \
			iter->second.crcs.end() != iter1; iter1++) {
			json_object * j_crc = json_object_new_int(*iter1);
			json_object_array_add(j_crcs, j_crc);
		}
		json_object_object_add(j_data_svr_md_record, "DATA_SVR_ID", j_data_svr_id);
		json_object_object_add(j_data_svr_md_record, "FILE_SIZE", j_file_size);
		json_object_object_add(j_data_svr_md_record, "CRC", j_crcs);
		json_object_array_add(j_data_svr_md_records, j_data_svr_md_record);
	}
	json_object_object_add(j, "CLIENT_MD_RECORD", j_client_md_record);
	json_object_object_add(j, "DATA_SVR_MD_RECORDS", j_data_svr_md_records);
	string buf = json_object_to_json_string(j);
	json_object_put(j);
	FILE * fp = fopen(file_path, "w+");
	if (NULL == fp) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fopen err, file:" << file_path << ", errno:" << errno);
		return -1;
	}
	int w = fwrite(buf.c_str(), buf.size(), 1, fp);
	fclose(fp);
	if (1 != w) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", fwrite error, file:" << file_path << ", content:" << buf << \
				", errno:" << errno);
		return -2;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, file:" << file_path << ", buf:" << buf);
	return 1;
}
	
int MetadataSvr::update_md_file_info_by_data_svr_notify(const char * file_path, int data_svr_id, vector<uint32_t> crcs, \
	int64_t file_size, int delete_flag) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", null file");
		return -1;
	}
	MD_FILE_INFO info;
	int ret = get_md_file_info(file_path, info);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get_md_file_info err, file:" << file_path << ", ret:" << ret);
		return -2;
	}
	if (1 == delete_flag) {
		info.data_svr_md_records.erase(data_svr_id);
	}
	else {
		map<int, DATA_SVR_MD_RECORD>::iterator iter = info.data_svr_md_records.find(data_svr_id);
		DATA_SVR_MD_RECORD data_svr_record;
		data_svr_record.file_size = file_size;
		data_svr_record.crcs = crcs; 
		if (info.data_svr_md_records.end() == iter) {
			info.data_svr_md_records.insert(map<int, DATA_SVR_MD_RECORD>::value_type(data_svr_id, data_svr_record));
		}
		else info.data_svr_md_records[data_svr_id] = data_svr_record;
	}
	ret = write_md_file_info(file_path, info);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", write_md_file_info error, file:" << file_path << ", ret:" << ret);
		return -3;
	}
	return 1;
}

int MetadataSvr::update_md_file_info_by_client_notify(const char * file_path, vector<uint32_t> crcs) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", null file");
		return -1;
	}
	MD_FILE_INFO info;
	int ret = get_md_file_info(file_path, info);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get_md_file_info err, file:" << file_path << ", ret:" << ret);
		return -2;
	}
	CLIENT_MD_RECORD r;
	r.crcs = crcs;
	info.client_md_record = r;
	ret = write_md_file_info(file_path, info);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", write_md_file_info error, file:" << file_path << ", ret:" << ret);
		return -3;
	}
	return 1;
}

int MetadataSvr::check_replicate(int32_t dir_id, int64_t file_id, MD_FILE_INFO info) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args, dir_id:" << dir_id << ", file_id:" << file_id);
		return -1;
	}
	int num_of_ok_replicas = 0;
	REPLICATE_TASK t;
	set<int> ok_ds_ids;
	for (map<int, DATA_SVR_MD_RECORD>::const_iterator iter = info.data_svr_md_records.begin(); \
		info.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, info.client_md_record.crcs)) {
			t.dir_id = dir_id;
			t.file_id = file_id;
			t.src_ds_id = iter->first;
			t.src_file_size = iter->second.file_size;
			t.src_crcs = iter->second.crcs;
			ok_ds_ids.insert(iter->first);
			num_of_ok_replicas++;
		}
	}
	if (num_of_ok_replicas >= _replicate_factor) {
		LOG4CXX_INFO(_logger, __FUNCTION__ << ", dir_id:" << dir_id << ", file_id:" << file_id << \
				", satisfy replicate factor:" << _replicate_factor << ", current ok replicas:" << \
				num_of_ok_replicas);
		return 1;
	}
	if (num_of_ok_replicas < 1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << " error, could not found ok replica");
		return -2;
	}
	string file_path = get_file_hash_path(_root_data_path.c_str(), dir_id, file_id);
	struct stat s;
	int ret = stat(file_path.c_str(), &s);
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not stat file:" << file_path << ", errno:" << errno);
		return -3;
	}
	int64_t now = time(NULL);
	if (now - s.st_mtime < _min_modify_interval_before_replicate) {
		LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", not in good time to replicate, dir_id:" << dir_id << \
				", file_id:" << file_id << ", seconds from last modify:" << now - s.st_mtime << \
				", min_modify_interval_before_replicate:" << _min_modify_interval_before_replicate);
		return 2;
	}
	int dst_ds_id = alloc_ds_4_replicate(ok_ds_ids);
	if (dst_ds_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", alloc_ds_4_replicate error, dir_id:" << dir_id << ", file_id:" << \
				file_id << ", ret:" << dst_ds_id);
		return -4;
	}
	t.dst_ds_id = dst_ds_id;
	put_replicate_task_to_queue(t);
	return 3;
}

int MetadataSvr::put_replicate_task_to_queue(REPLICATE_TASK task) {
	if (task.dir_id < 0 || task.file_id < 0 || task.src_ds_id < 0 || task.src_file_size < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args, dir_id:" << task.dir_id << ", file_id:" << \
				task.file_id << ", src_ds_id:" << task.src_ds_id << ", src_file_size:" << task.src_file_size);
		return -1;
	}
	time_t t = time(NULL);
	struct tm now; 
	localtime_r(&t, &now);
	int replicate_waiting_response_queue_id = now.tm_hour / 8;
	pthread_mutex_lock(&_replicate_tasks_mutex);
	if (_replicate_waiting_notify_tasks.find(task) == _replicate_waiting_notify_tasks.end() && \
			_replicate_waiting_response_tasks_0.find(task) == _replicate_waiting_response_tasks_0.end() && \
			_replicate_waiting_response_tasks_1.find(task) == _replicate_waiting_response_tasks_1.end() && \
			_replicate_waiting_response_tasks_2.find(task) == _replicate_waiting_response_tasks_2.end()) {
		_replicate_waiting_notify_tasks.insert(task);
		if (0 == replicate_waiting_response_queue_id)  {
			_replicate_waiting_response_tasks_1.clear();
			_replicate_waiting_response_tasks_0.insert(task);
		}
		else if (1 == replicate_waiting_response_queue_id) {
				_replicate_waiting_response_tasks_2.clear();
				_replicate_waiting_response_tasks_1.insert(task);
		}
		else {
			_replicate_waiting_response_tasks_0.clear();
			_replicate_waiting_response_tasks_2.insert(task);
		}
		pthread_cond_signal(&_replicate_tasks_cond);
		pthread_mutex_unlock(&_replicate_tasks_mutex);
		return 1;
	}
	pthread_mutex_unlock(&_replicate_tasks_mutex);
        return -2;
}

int MetadataSvr::get_replicate_task_from_waiting_notify_queue(REPLICATE_TASK & task) {
	pthread_mutex_lock(&_replicate_tasks_mutex);
        pthread_cond_wait(&_replicate_tasks_cond, &_replicate_tasks_mutex);
        if (_replicate_waiting_notify_tasks.size() < 1) {
                pthread_mutex_unlock(&_replicate_tasks_mutex);
                return -1;
        }
        set<REPLICATE_TASK>::const_iterator iter = _replicate_waiting_notify_tasks.begin();
        task = *iter;
        _replicate_waiting_notify_tasks.erase(task);
        pthread_mutex_unlock(&_replicate_tasks_mutex);
        return 1;
}

int MetadataSvr::erase_replicate_task_from_waiting_response_queue(REPLICATE_TASK task) {
	pthread_mutex_lock(&_replicate_tasks_mutex);
	_replicate_waiting_response_tasks_0.erase(task);
	_replicate_waiting_response_tasks_1.erase(task);
	_replicate_waiting_response_tasks_2.erase(task);
        pthread_mutex_unlock(&_replicate_tasks_mutex);
}

void * MetadataSvr::notify_replicate_thread_routine(void * arg) {
	MetadataSvr * mds = (MetadataSvr *)arg;
	while (true) {
		REPLICATE_TASK t;
		int ret = mds->get_replicate_task_from_waiting_notify_queue(t);
		if (ret < 0) continue;
		DataSvrClient * dsc = mds->_dscg.get_data_svr_client(t.dst_ds_id);
        	if (NULL == dsc) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get data svr client, dst_ds_id:" << t.dst_ds_id);
			mds->erase_replicate_task_from_waiting_response_queue(t);
		}
		ret = dsc->replicate(t.dir_id, t.file_id, t.src_ds_id, t.src_file_size, t.src_crcs);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", notify ds replicate error, dst_ds_id:" << t.dst_ds_id << \
					", src_ds_id:" << t.src_ds_id << ", dir_id:" << t.dir_id << ", file_id:" << t.file_id << \
					", src_file_size:" << t.src_file_size);
			mds->erase_replicate_task_from_waiting_response_queue(t);
		}
		else {
			LOG4CXX_INFO(_logger, __FUNCTION__ << ", notify ds replicate ok, dst_ds_id:" << t.dst_ds_id << \
					", src_ds_id:" << t.src_ds_id << ", dir_id:" << t.dir_id << ", file_id:" << t.file_id << \
					", src_file_size:" << t.src_file_size);
	
		}
        }
}

int MetadataSvr::alloc_ds_4_replicate(set<int> ok_replica_ids) {
	list<DS_CNF> l_ds;
        int ret = _csc.get_all_ds_cnf(l_ds);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get all ds cnf error, ret:" << ret);
                return -1;
        }
	list<DS_CNF> l_candidate_ds;
	int total_weight = 0;
	for (list<DS_CNF>::const_iterator iter = l_ds.begin(); l_ds.end() != iter; iter++) {
		if (ok_replica_ids.find(iter->id) != ok_replica_ids.end()) continue;
		if (DS_STATUS_OK != iter->status) continue;
		if (iter->write_weight < 1) continue;
		if (iter->mds_id != _my_mds_id) continue;
		l_candidate_ds.push_back(*iter);
		total_weight += iter->write_weight;
	}
	int r = rand() % total_weight;
	int x = 0;
	for(list<DS_CNF>::const_iterator iter = l_candidate_ds.begin(); l_candidate_ds.end() != iter; iter++) {
		if (iter->write_weight >= x) return iter->id;
		x += iter->write_weight;
	}
	LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not found appropriate ds for replicate, total write weight:" << \
			total_weight);
	return -2;
}

bool MetadataSvr::is_crcs_equal(vector<uint32_t> crcs1, vector<uint32_t> crcs2) {
        if (crcs1.size() != crcs2.size()) return false;
        int i = 0;
        for (vector<uint32_t>::const_iterator iter = crcs1.begin(); crcs1.end() != iter; iter++) {
                if (*iter != crcs2[i]) return false; 
                i++;
        }
        return true;
}

int MetadataSvr::acquire_write_md_file_lock(int32_t dir_id, int64_t file_id) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	int t = dir_id % WRITE_MD_FILE_MUTEX_NUM;
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << " try, dir_id:" << dir_id << ", file_id:" << file_id << ", t:" << t);
	pthread_mutex_lock(&_write_md_file_mutexes[t]);
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << " ok, dir_id:" << dir_id << ", file_id:" << file_id << ", t:" << t);
	return 1;
}

int MetadataSvr::release_write_md_file_lock(int32_t dir_id, int64_t file_id) {
	if (dir_id < 0 || file_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	int t = dir_id % WRITE_MD_FILE_MUTEX_NUM;
	pthread_mutex_unlock(&_write_md_file_mutexes[t]);
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", dir_id:" << dir_id << ", file_id:" << file_id << ", t:" << t);
	return 1;
}
